package net.jacqeush.mini.server.hander;

import com.aliyun.opensearch.sdk.dependencies.com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import net.jacqeush.mini.rpc.codec.model.Request;
import net.jacqeush.mini.rpc.codec.model.Response;
import net.sf.cglib.reflect.FastClass;
import net.sf.cglib.reflect.FastMethod;

import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.concurrent.*;

/**
 * @author Jacquesh
 * @version 1.0
 * @date 2020/10/17 2:31
 */
@Slf4j
public class SeverHandler extends SimpleChannelInboundHandler<Request> {

    ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("socket-connect-pool-%d").build();
    ExecutorService executor = new ThreadPoolExecutor(16, 16,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(Integer.MAX_VALUE),
            namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    HashMap<String, Object> handlerMaps=new HashMap<String, Object>();
    public SeverHandler(HashMap<String, Object> handlerMaps) {
        this.handlerMaps=handlerMaps;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Request request) throws Exception {
        log.info("接收到请求信息[{}]",request);
        executor.submit(()->{
                Response response=new Response();
                response.setRequestId(request.getRequestId());
            try {
                Object result = handle(request);
                response.setResult(result);
            }catch (Throwable e){
                response.setThrowable(e);
                log.error("rpc service handle request Throwable: " + e);
            }

            channelHandlerContext.writeAndFlush(response).addListener((future) ->{
                    if(future.isSuccess()) {
                        //TODO 后置处理器
                        log.info("返回处理信息[{}]",response);
                     log.info("后置处理器");
                    }

            });
        });
    }

    private Object handle(Request request) throws InvocationTargetException {
        String className = request.getClassName();
        Object serviceRef = handlerMaps.get(className);
        Class<?> serviceClass = serviceRef.getClass();
        String methodName = request.getMethodName();
        Class<?>[] paramsTypes = request.getParamsType();
        Object[] params = request.getParams();

        // JDK invoke

        // Cglib
        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, paramsTypes);
        return serviceFastMethod.invoke(serviceRef, params);
    }

    /**
     * $ netty 异常
     * @param ctx 上下文
     * @param cause 原因
     *
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  {
        log.error("channel exception by cause"+cause);
        ctx.close();
    }
}
